package top.pcat.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import top.pcat.config.KafkaConfig;
import top.pcat.config.ServerConfig;
import top.pcat.service.MessagingService;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

@Slf4j
public class KafkaConsumerUtil {

    private static volatile KafkaConsumer<String, byte[]> consumer; // Kafka 消费者

    public KafkaConsumerUtil(KafkaConfig kafkaConfig) {
        if (consumer == null) {
            synchronized (KafkaConsumerUtil.class) {
                if (consumer == null) {
                    Properties props = new Properties();
                    props.put("bootstrap.servers", kafkaConfig.getBootstrapServers());

                    // 配置 Kafka 消费者
                    props.put("group.id", ServerConfig.serverId);
                    props.put("key.deserializer", kafkaConfig.getConsumerKeyDeserializer());
                    props.put("value.deserializer", kafkaConfig.getConsumerValueDeserializer());
                    consumer = new KafkaConsumer<>(props);
                }
            }
        }
    }

    /**
     * 订阅 Kafka 话题并接收消息
     *
     * @param topic 话题名称
     */
    public void receiveMsgByKafka(String topic) {
        consumer.subscribe(Collections.singleton(topic));
        MessagingService messagingService = new MessagingService();
        while (true) {
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(2000));
            for (ConsumerRecord<String, byte[]> record : records) {
                // 调用消息处理
                log.info("收到的kafka消息"+record.key()+" value:"+record.value());
                messagingService.processMsgReceivedByKafka(record.value());
            }
        }
    }
}
